package com.distributer.delayqueue.delayqueue.event;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Date;
import java.util.List;

import javax.sql.DataSource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.distributer.delayqueue.job.JobMsg;
import com.distributer.delayqueue.job.JobWrapp;
import com.google.common.collect.Lists;

public class RdbStore {

	public static final Logger LOGGER = LoggerFactory.getLogger(RdbStore.class);

	private DataSource dataSource;

	public RdbStore(final DataSource dataSource) {
		this.dataSource = dataSource;
	}

	public boolean insertJob(JobMsg jobMsg) {
		boolean result = false;
		String sql = "INSERT INTO delay_queue_job (job_Id, job_key, job_topic, body, delay_time, status, bucket_name, create_time,update_time) VALUES (?,?, ?, ?, ?, ?, ?, ?, ?)";
		try (Connection conn = dataSource.getConnection(); PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
			preparedStatement.setString(1, jobMsg.getJobId());
			preparedStatement.setString(2, jobMsg.getJobKey());
			preparedStatement.setString(3, jobMsg.getJobTopic());
			preparedStatement.setString(4, jobMsg.getBody());
			preparedStatement.setLong(5, jobMsg.getDelayTime());
			preparedStatement.setInt(6, jobMsg.getStatus());
			if (jobMsg instanceof JobWrapp) {
				preparedStatement.setString(7, ((JobWrapp) jobMsg).getBucketName());
			}
			preparedStatement.setTimestamp(8, new Timestamp(jobMsg.getCreateTime().getTime()));
			preparedStatement.setTimestamp(9, new Timestamp(jobMsg.getCreateTime().getTime()));
			preparedStatement.execute();
			insertLogJob(jobMsg, conn);
			result = true;
		} catch (final SQLException ex) {
			LOGGER.error("INSERT数据发生错误", ex);
		}
		return result;
	}

	public boolean updateJobsStatus(JobMsg jobMsg) {
		boolean result = false;
		String sql = "UPDATE delay_queue_job SET status =? , update_time = ? WHERE job_Id=? ";
		try (Connection conn = dataSource.getConnection(); PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
			preparedStatement.setInt(1, jobMsg.getStatus());
			preparedStatement.setTimestamp(2, new Timestamp(new Date().getTime()));
			preparedStatement.setString(3, jobMsg.getJobId());
			preparedStatement.execute();
			insertLogJob(jobMsg, conn);
			result = true;
		} catch (final SQLException ex) {
			LOGGER.error("UPDATE数据发生错误", ex);
		} finally {

		}
		return result;
	}

	/**
	 * 添加运行规矩日志 注意 这个方法不会自动关闭数据库链接
	 */
	public boolean insertLogJob(JobMsg jobMsg, Connection conn) {
		boolean result = false;
		String sql = "INSERT INTO delay_queue_job_log (id, status, thread, update_time) VALUES (?, ?, ?, ?);";
		try {
			PreparedStatement preparedStatement = conn.prepareStatement(sql);
			preparedStatement.setString(1, jobMsg.getJobId());
			preparedStatement.setInt(2, jobMsg.getStatus());
			preparedStatement.setString(3, Thread.currentThread().getName());
			preparedStatement.setTimestamp(4, new Timestamp(new Date().getTime()));
			preparedStatement.execute();
			result = true;
		} catch (final SQLException ex) {
			LOGGER.error("INSERT日志数据发生错误", ex);
		}
		return result;
	}

	/**
	 * 获取待恢复的数据行数 这里是获取 不是已完成 并且不是已删除的数据
	 */
	public int getNotFinshDataCount() {
		String countSql = "SELECT  count(1) FROM delay_queue_job  where status <> 3 and status <> 4 ";
		int count = 0;
		try (Connection conn = dataSource.getConnection(); PreparedStatement preparedStatement = conn.prepareStatement(countSql)) {
			ResultSet rs = preparedStatement.executeQuery();
			if (rs != null && rs.next()) {
				count = rs.getInt(1);
			}
		} catch (final SQLException ex) {
			LOGGER.error("COUNT数据发生错误", ex);
		}
		return count;
	}

	public JobMsg getJobMyId(String jobId) {
		String sql = "SELECT job_Id,job_key,job_topic,body,delay_time,status,bucket_name,create_time FROM delay_queue_job WHERE job_Id = ?";
		try (Connection conn = dataSource.getConnection(); PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
			preparedStatement.setString(1, jobId);
			ResultSet rs = preparedStatement.executeQuery();
			while (rs != null && rs.next()) {
				JobWrapp msg = new JobWrapp();
				msg.setJobId(rs.getString("job_Id"));
				msg.setJobKey(rs.getString("job_key"));
				msg.setJobTopic(rs.getString("job_topic"));
				msg.setBody(rs.getString("body"));
				msg.setDelayTime(rs.getLong("delay_time"));
				msg.setStatus(rs.getInt("status"));
				msg.setBucketName(rs.getString("bucket_name"));
				msg.setCreateTime(rs.getDate("create_time"));
				return msg;
			}
		} catch (final SQLException ex) {
			LOGGER.error("getJobMyId数据发生错误", ex);
		}
		return null;
	}

	public List<JobMsg> getNotFinshDataList(int start, int size) {
		String sql = "SELECT job_Id,job_key,job_topic,body,delay_time,status,bucket_name,create_time FROM delay_queue_job where "
				+ "where status <> 3 and status <> 4 order by create_time ,delay limit ?, ?";
		List<JobMsg> msgList = Lists.newArrayList();
		try (Connection conn = dataSource.getConnection(); PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
			preparedStatement.setInt(1, start);
			preparedStatement.setInt(2, size);
			ResultSet rs = preparedStatement.executeQuery();
			while (rs != null && rs.next()) {
				JobWrapp msg = new JobWrapp();
				msg.setJobId(rs.getString("job_Id"));
				msg.setJobKey(rs.getString("job_key"));
				msg.setJobTopic(rs.getString("job_topic"));
				msg.setBody(rs.getString("body"));
				msg.setDelayTime(rs.getLong("delay_time"));
				msg.setStatus(rs.getInt("status"));
				msg.setBucketName(rs.getString("bucket_name"));
				msg.setCreateTime(rs.getDate("create_time"));
				msgList.add(msg);
			}
		} catch (final SQLException ex) {
			LOGGER.error("getNotFinshDataList数据发生错误", ex);
		} finally {

		}
		return msgList;
	}
}
